package com.xjgz.aio.socket.enhance;

import java.io.IOException;
import java.net.SocketAddress;
import java.net.SocketOption;
import java.nio.channels.AcceptPendingException;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.CompletionHandler;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;
import java.util.concurrent.Future;

public class EnhanceAsynchronousServerSocketChannel extends AsynchronousServerSocketChannel {

    /**
     * 底层的服务器Socket通道，用于实际的网络IO操作
     */
    private final ServerSocketChannel serverSocketChannel;

    /**
     * 异步通道组，用于管理通道的线程资源和事件分发
     */
    private final EnhanceAsynchronousChannelGroup enhanceAsynchronousChannelGroup;

    /**
     * 接受连接的回调处理器，用于处理新连接建立后的回调逻辑
     */
    private CompletionHandler<AsynchronousSocketChannel, Object> acceptCompletionHandler;

    /**
     * 用于Future方式调用时的回调处理器
     */
    private FutureCompletionHandler<AsynchronousSocketChannel, Void> acceptFuture;

    /**
     * 接受连接操作的附加对象，可以在回调传递额外的数据
     */
    private Object attachment;

    /**
     * 用于接受连接操作的选择键，管理通道的接受事件注册
     */
    private SelectionKey selectionKey;

    /**
     * 标识是否有待处理的接受连接操作
     */
    private boolean acceptPending;

    /**
     * 低内存模式
     */
    private final boolean lowMemory;

    /**
     * 接受连接操作的调用计数器
     */
    private int acceptInvoker;

    public EnhanceAsynchronousServerSocketChannel(
        EnhanceAsynchronousChannelGroup enhanceAsynchronousChannelGroup,
        boolean lowMemory
    ) throws IOException {
        super(enhanceAsynchronousChannelGroup.provider());
        this.enhanceAsynchronousChannelGroup = enhanceAsynchronousChannelGroup;
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        this.lowMemory = lowMemory;
    }

    @Override
    public AsynchronousServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
        serverSocketChannel.bind(local, backlog);
        return this;
    }

    @Override
    public Future<AsynchronousSocketChannel> accept() {
        FutureCompletionHandler<AsynchronousSocketChannel, Void> acceptFuture = new FutureCompletionHandler<>();
        accept(null, acceptFuture);
        this.acceptFuture = acceptFuture;
        return acceptFuture;
    }

    @SuppressWarnings("unchecked")
    @Override
    public <A> void accept(A attachment, CompletionHandler<AsynchronousSocketChannel, ? super A> handler) {
        if (acceptPending) {
            throw new AcceptPendingException();
        }
        acceptPending = true;
        this.acceptCompletionHandler = (CompletionHandler<AsynchronousSocketChannel, Object>) handler;
        this.attachment = attachment;
        doAccept();
    }

    public void doAccept() {
        try {
            if (acceptFuture != null && acceptFuture.isDone()) {
                resetAccpet();
                EnhanceAsynchronousChannelGroup.removeOps(selectionKey, SelectionKey.OP_ACCEPT);
                return;
            }
            SocketChannel socketChannel = null;
            if (acceptInvoker++ < EnhanceAsynchronousChannelGroup.MAX_INVOKER) {
                socketChannel = serverSocketChannel.accept();
            }
            if (socketChannel != null) {
                EnhanceAsynchronousSocketChannel asynchronousSocketChannel =
                    new EnhanceAsynchronousSocketChannel(enhanceAsynchronousChannelGroup, socketChannel, lowMemory);
                socketChannel.configureBlocking(false);
                socketChannel.finishConnect();
                CompletionHandler<AsynchronousSocketChannel, Object> completionHandler = acceptCompletionHandler;
                Object attach = attachment;
                resetAccpet();
                completionHandler.completed(asynchronousSocketChannel, attach);
                if (!acceptPending && selectionKey != null) {
                    EnhanceAsynchronousChannelGroup.removeOps(selectionKey, SelectionKey.OP_ACCEPT);
                }
            }
            // 首次注册Selector
            else if (selectionKey == null) {
                enhanceAsynchronousChannelGroup.commnoWorker.addRegister(selector -> {
                    try {
                        selectionKey = serverSocketChannel
                            .register(selector, SelectionKey.OP_ACCEPT, EnhanceAsynchronousServerSocketChannel.this);
                    } catch (ClosedChannelException e) {
                        acceptCompletionHandler.failed(e, attachment);
                    }
                });
            }
        } catch (IOException e) {
            this.acceptCompletionHandler.failed(e, attachment);
        } finally {
            acceptInvoker = 0;
        }
    }

    private void resetAccpet() {
        acceptPending = false;
        acceptFuture = null;
        acceptCompletionHandler = null;
        attachment = null;
    }

    @Override
    public SocketAddress getLocalAddress() throws IOException {
        return serverSocketChannel.getLocalAddress();
    }

    @Override
    public <T> AsynchronousServerSocketChannel setOption(SocketOption<T> name, T value) throws IOException {
        serverSocketChannel.setOption(name, value);
        return this;
    }

    @Override
    public void close() throws IOException {
        serverSocketChannel.close();
    }

    @Override
    public boolean isOpen() {
        return serverSocketChannel.isOpen();
    }

    @Override
    public <T> T getOption(SocketOption<T> name) throws IOException {
        return serverSocketChannel.getOption(name);
    }

    @Override
    public Set<SocketOption<?>> supportedOptions() {
        return serverSocketChannel.supportedOptions();
    }

}
